package com.ctliu.datastructure.blockingqueue;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * 单锁实现
 * @param <E> 元素类型
 */
@SuppressWarnings("all")
public class BlockingQueue1<E> implements BlockingQueue<E> {

    private final E[] array;
    private int head;// 记录开始的元素索引
    private int tail;// 记录结束的元素索引
    private int size;// 辅助判断队列空还是满

    public BlockingQueue1(int capacity) {
        array = (E[]) new Object[capacity];
    }

    private ReentrantLock lock = new ReentrantLock();//锁对象

//    条件变量对象(只能通过锁对象创建)
//    是一个(消费者/生产者)集合
    private Condition headWaits = lock.newCondition();// 等待向尾添加对象的线程 配合poll()方法使用
    private Condition tailWaits = lock.newCondition();// 等着删除头部对象的线程 配合offer()方法使用

//    判空
    private boolean isEmpty() {
        return size == 0;
    }
//    判满
    private boolean isFull() {
        return size == array.length;
    }

    @Override
    public void offer(E e) throws InterruptedException {
        lock.lockInterruptibly();// 加锁(可以在阻塞状态下被打断)
        try {
            // 这里必须是while 不能是if! 防止虚假唤醒
            while (isFull()) {
//                队列满了
                tailWaits.await(); // offer线程暂停(让这个线程进入阻塞状态) signal()方法是恢复运行
            }
            array[tail] = e;
//            维护tail尾指针
            if (++tail == array.length) {
//                循环队列(如果达到长度,置位0,保证有效索引访问数组)
                tail = 0;
            }
            size++;// 维护size
            headWaits.signal();// 唤醒阻塞中的poll()线程
        } finally {
//            固定套路,解锁放在finally里,保证一定执行
            lock.unlock();
        }
    }


//    超时,返回失败(有限时间的等待)
    @Override
    public boolean offer(E e, long timeout) throws InterruptedException { // 毫秒
        lock.lockInterruptibly();
        try {
            //毫秒 -> 纳秒  利用TimeUnit类中的毫秒对象,转换为纳秒
            long t = TimeUnit.MILLISECONDS.toNanos(timeout);
            while (isFull()) {
                if (t <= 0) {
//              时间不够,返回
                    return false;
                }
//                更新t,防止一直等待
                t = tailWaits.awaitNanos(t); // 最多等待多少 (单位纳秒)  返回值代表剩余时间
            }
            array[tail] = e;
            if (++tail == array.length) {
                tail = 0;
            }
            size++;
            headWaits.signal();
            return true;
        } finally {
            lock.unlock();
        }
    }

//    poll方法
    @Override
    public E poll() throws InterruptedException {
//        加锁
        lock.lockInterruptibly();
        try {
            while (isEmpty()) {
//                如果未空,让当前线程等待(放入headWaits集合中)
                headWaits.await();
            }
//            取走头元素
            E e = array[head];
            array[head] = null; // help GC
            if (++head == array.length) {
                head = 0;
            }
            size--;
            tailWaits.signal();// 通知等待不满的offer线程
            return e;
        } finally {
            lock.unlock();
        }
    }

    @Override
    public String toString() {
        return Arrays.toString(array);
    }

    public static void main(String[] args) throws InterruptedException {
        BlockingQueue1<String> queue = new BlockingQueue1<>(3);
        queue.offer("任务1");

        new Thread(()->{
            try {
                queue.offer("任务2");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "offer").start();

        new Thread(()->{
            try {
                System.out.println(queue.poll());
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }, "poll").start();
    }
}
